/*
 *  TazioTransaction.java        22-ott-2009
 *   
 * Copyright (c) 2009 mentalsmash.org
 *  
 *  Licensed under the Apache License, Version 2.0 (the "License");
 *  you may not use this file except in compliance with the License.
 *  You may obtain a copy of the License at
 * 
 *  http://www.apache.org/licenses/LICENSE-2.0
 * 
 *  Unless required by applicable law or agreed to in writing, software
 *  distributed under the License is distributed on an "AS IS" BASIS,
 *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *  See the License for the specific language governing permissions and
 *  limitations under the License.
 * 
 *  The use of the Apache License does not indicate that this project is
 *  affiliated with the Apache Software Foundation.
 */

package org.mentalsmash.tazio.transactions.control.impl;

import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.Future;

import org.mentalsmash.tazio.transactions.control.Snapshooter;
import org.mentalsmash.tazio.transactions.control.SystemTransactionManager;
import org.mentalsmash.tazio.transactions.control.TazioTransaction;
import org.mentalsmash.tazio.transactions.control.TransactionStateException;
import org.mentalsmash.tazio.transactions.control.TxStateObserver;
import org.mentalsmash.tazio.transactions.model.IllegalSeqNoException;
import org.mentalsmash.tazio.transactions.model.RootSet;
import org.mentalsmash.tazio.transactions.model.Snapshot;
import org.mentalsmash.tazio.transactions.model.TransactionalObject;
import org.mentalsmash.tazio.transactions.model.TransactionalOperation;
import org.mentalsmash.tazio.transactions.model.identifiers.OID;
import org.mentalsmash.tazio.transactions.model.identifiers.TID;
import org.mentalsmash.tazio.transactions.user.RollbackException;
import org.mentalsmash.tazio.transactions.user.Transaction;
import org.mentalsmash.tazio.transactions.user.TransactionStatus;
import org.mentalsmash.tazio.transactions.user.TransactionalSystemException;
import org.mentalsmash.tazio.transactions.user.TransactionalSystemRuntimeException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
import com.google.inject.Inject;

public abstract class TazioTransactionImpl implements Transaction,
	TazioTransaction {

    private static final long serialVersionUID = -1599381999311796144L;

    private transient final static Logger log = LoggerFactory
	    .getLogger(TazioTransactionImpl.class);

    /**
     * The instance used to represent the absence of transactions in places
     * where the <code>null</code> value could generate ambiguities.
     * <p>
     * Its TID is a bunch of zeros
     * </p>
     */
    public final static TazioTransactionImpl NULL_TRANSACTION = NullTazioTransaction
	    .getInstance();

    private final TID transactionID;

    private long seqNo = INITIAL_SEQNO; // initialized to an invalid seqno

    private TransactionStatus status = TransactionStatus.NO_TRANSACTION;

    // snapshot are only used locally to achieve isolation in transactions
    private final transient Map<OID, Snapshot> snapshots = Maps
	    .newHashMap();
    

    // map of operations performed on transactional objects by this transaction.
    // It is sent to the RM cluster for commitment
    // along with the write set and read set
    private final Map<OID, List<TransactionalOperation>> operations = Maps
	    .newHashMap();

    private final Set<OID> writeSet = Sets.newHashSet();
    private final Set<OID> readSet = Sets.newHashSet();

    // internally synchronized. Remember to synchronize over iterators
    private final transient Set<TxStateObserver> stateObservers;

    // responsible for taking snapshot of transactional objects
    private final transient Snapshooter snapshooter;

    // if true the given transaction can do nothing then rollback
    private boolean rollBackOnly = false;

    // the transaction manager instance managing this transaction
    private transient final SystemTransactionManager transactionManager;

    @Inject
    protected TazioTransactionImpl(final SystemTransactionManager tm,
	    final Snapshooter snapshooter, final TID tid) {
	final Set<TxStateObserver> unsyncObservers = Sets.newHashSet();
	this.stateObservers = Collections.synchronizedSet(unsyncObservers);
	this.transactionID = tid;
	this.transactionManager = tm;
	this.snapshooter = snapshooter;

    }

    /*
     * (non-Javadoc)
     * 
     * @seeorg.mentalsmash.tazio.transactions.control.impl.TazioTransaction#
     * registerStateObserver
     * (org.mentalsmash.tazio.transactions.control.TxStateObserver)
     */
    public void registerStateObserver(final TxStateObserver observer) {
	log.debug("Adding a state observer for tx {} : {}", this.transactionID,
		observer);
	this.stateObservers.add(observer);
    }

    /*
     * (non-Javadoc)
     * 
     * @seeorg.mentalsmash.tazio.transactions.control.impl.TazioTransaction#
     * removeStateObserver
     * (org.mentalsmash.tazio.transactions.control.TxStateObserver)
     */
    public boolean removeStateObserver(final TxStateObserver observer) {
	log.debug("Removing a state obeserver for tx {} : {}",
		this.transactionID, observer);
	return this.stateObservers.remove(observer);
    }

    /*
     * (non-Javadoc)
     * 
     * @seeorg.mentalsmash.tazio.transactions.control.impl.TazioTransaction#
     * getTransactionId()
     */
    @Override
    public TID getTransactionId() {
	return this.transactionID;
    }

    /*
     * (non-Javadoc)
     * 
     * @see
     * org.mentalsmash.tazio.transactions.control.impl.TazioTransaction#getSeqNo
     * ()
     */
    public long getSeqNo() {
	return this.seqNo;
    }

    /**
     * Sets the sequence number of this transaction to the given value
     * 
     * @param seqNo
     *            the new sequece number
     * @throws IllegalSeqNoException
     *             if the provided sequence number is negative
     */
    @Override
    public void setSeqNo(final long seqNo) {
	if (seqNo < 0) {
	    log.error("Someone tried to set transaction's seqno to {}", seqNo);
	    throw new IllegalSeqNoException(
		    "Someone tried to set transaction's seqno to " + seqNo);
	}
	this.seqNo = seqNo;
    }

    /*
     * (non-Javadoc)
     * 
     * @see
     * org.mentalsmash.tazio.transactions.control.impl.TazioTransaction#getStatus
     * ()
     */
    @Override
    public TransactionStatus getStatus() {
	return this.status;
    }

    /**
     * Changes the internal status of the <tt>TazioTransaction</tt>. This method
     * does not make a complete state transitions, but simply sets the internal
     * variable.
     * <p>
     * <strong>Consider</strong> using
     * {@link #stateTransition(TransactionStatus)} instead
     * </p>
     * 
     * @param newStatus
     *            the new status to set
     * @see #stateTransition(TransactionStatus)
     */
    protected void setStatus(final TransactionStatus newStatus) {
	this.status = newStatus;
    }

    /*
     * (non-Javadoc)
     * 
     * @see
     * org.mentalsmash.tazio.transactions.control.impl.TazioTransaction#isActive
     * ()
     */
    public boolean isActive() {
	final boolean notActive = this.getStatus().equals(
		TransactionStatus.NO_TRANSACTION)
		|| this.getStatus().equals(TransactionStatus.COMMITTED)
		|| this.getStatus().equals(TransactionStatus.ROLLED_BACK);
	return !notActive;
    }

    /**
     * Changes the state of the <tt>TazioTransaction</tt> to the given
     * {@link TransactionStatus} and notifies the registered
     * {@link TxStateObserver}s
     * 
     * @param to
     *            the destination status to switch to
     */
    @Override
    public void stateTransition(final TransactionStatus to) {
	if (!to.equals(this.status)) { // avoids useless state transitions and
	    // notifications
	    final TransactionStatus from = this.status;
	    this.status = to;
	    synchronized (this.stateObservers) {
		for (final TxStateObserver observer : this.stateObservers) {
		    observer.statusChanged(from, to, this.getTransactionId());
		}
	    }
	}
    }

    /*
     * (non-Javadoc)
     * 
     * @see
     * org.mentalsmash.tazio.transactions.control.impl.TazioTransaction#getReadSet
     * ()
     */
    public Set<OID> getReadSet() {
	return Collections.unmodifiableSet(this.readSet);
    }

    /*
     * (non-Javadoc)
     * 
     * @see
     * org.mentalsmash.tazio.transactions.control.impl.TazioTransaction#getWriteSet
     * ()
     */
    public Set<OID> getWriteSet() {
	return Collections.unmodifiableSet(this.writeSet);
    }

    /**
     * Adds an oid to the write set, indicating that the corresponding object
     * was written in the context of this transaction
     * 
     * @param oid
     *            the {@link OID} of the object to add
     */
    @Override
    public void addToWriteSet(final OID oid) {
	this.writeSet.add(oid);
    }

    /**
     * Adds an oid to the read set, indicating that the corresponding object was
     * read in the context of this transaction
     * 
     * @param oid
     *            the {@link OID} of the object to add
     */
    @Override
    public void addToReadSet(final OID oid) {
	this.readSet.add(oid);
    }

    /**
     * Returns the private snapshot for a transactional object, or
     * <code>null</code> if there is no snapshot for such an object
     * 
     * @param objectId
     *            the {@link OID} instance identifying a transactional object
     * @return a private snapshot of the object identified by
     *         <code>objectId</code> or <code>null</cod> if no such an object
     */
    @Override
    public Snapshot retrieveSnapshot(final OID objectId) {
	return this.snapshots.get(objectId);
    }

    /**
     * Returns an unmodifiable view of a map containing the snapshot taken by
     * this transaction
     * 
     * @return an unmodifiable view of a map containing the snapshot taken by
     *         this transaction
     */
    protected Map<OID, Snapshot> getSnapshots() {
	return Collections.unmodifiableMap(this.snapshots);
    }

    /**
     * Returns the list of operations performed by this transaction on a given
     * transactional object, or <code>null</code> if no operation was performed
     * on the given object
     * <p>
     * <strong>BEWARE:</strong> the methods returns <code>null</code> even if no
     * object having the desired {@link OID} exists
     * </p>
     * 
     * @return he list of operations performed by this transaction on a given
     *         transactional object, or <code>null</code> if no operation was
     *         performed on the given object
     */
    @Override
    public List<TransactionalOperation> getOperations(final OID objectId) {
	return this.operations.get(objectId);
    }

    /**
     * returns the map of every operation performed by this transaction.
     * <p>
     * Changing this map, <strong>does affect<strong> the state of the
     * transaction.
     * </p>
     * 
     * @return
     */
    protected Map<OID, List<TransactionalOperation>> getAllOperations() {
	return this.operations;
    }

    /**
     * Log an operation on the transaction
     * 
     * @param op
     *            the operation instance to log
     */
    @Override
    public void logOperation(final TransactionalOperation op) {
	final OID objectId = op.getTargetObject();
	List<TransactionalOperation> operationsOnObj = this.operations
		.get(objectId);
	if (operationsOnObj == null) {
	    operationsOnObj = Lists.newArrayList();
	    this.operations.put(objectId, operationsOnObj);
	}
	operationsOnObj.add(op);
    }

    /**
     * Takes a snapshot of the given {@link TransactionalObject} and adds it to
     * the private snapshots of this transaction
     * 
     * @param obj
     *            the {@link TransactionalObject} instance to take the snapshot
     *            of
     */
    @Override
    public void takeSnapshot(final TransactionalObject obj) {
	this.snapshooter.createSnapshot(obj, this.snapshots);
    }

    /* ************************************************************************ */
    /* From org.mentalsmash.tazio.transactions.Transaction interface */
    /* ************************************************************************ */

    /*
     * (non-Javadoc)
     * 
     * @see
     * org.mentalsmash.tazio.transactions.control.impl.TazioTransaction#commit()
     */
    @Override
    public void commit() throws RollbackException, TransactionStateException,
	    TransactionalSystemException {
	if(!this.getStatus().equals(TransactionStatus.RUNNING)){
	    throw new TransactionStateException("Cannot commit a not running transaction");
	}
	
	
	if (!threadOwnsTransaction()) {
	    log
		    .error("A thread has tried to commit a transaction it does not own");
	    throw new TransactionalSystemRuntimeException(
		    "Thread does not own the transaction");
	}
	this.transactionManager.commit();
    }

    /**
     * Checks whether this instance of <tt>TazioTransaction</tt> is the same
     * transaction that is currently associated to the running thread
     * 
     * @return <code>true</code> if this instance is the one associated to the
     *         current thread
     */
    private boolean threadOwnsTransaction() {
	return this.equals(this.transactionManager.getTransaction());
    }

    /*
     * (non-Javadoc)
     * 
     * @see
     * org.mentalsmash.tazio.transactions.control.impl.TazioTransaction#rollback
     * ()
     */
    @Override
    public void rollback() throws TransactionStateException {
	if(!this.getStatus().equals(TransactionStatus.RUNNING)){
	    throw new TransactionStateException("Cannot rollback a not running transaction");
	}
	if (!threadOwnsTransaction()) {
	    log
		    .error("A thread has tried to commit a transaction it does not own");
	    throw new TransactionalSystemRuntimeException(
		    "Thread does not own the transaction");
	}
	this.transactionManager.rollback();
    }

    /*
     * (non-Javadoc)
     * 
     * @seeorg.mentalsmash.tazio.transactions.control.impl.TazioTransaction#
     * setRollbackOnly()
     */
    @Override
    public void setRollbackOnly() {
	this.rollBackOnly = true;
    }

    /*
     * (non-Javadoc)
     * 
     * @see
     * org.mentalsmash.tazio.transactions.control.impl.TazioTransaction#commitAsync
     * ()
     */
    @Override
    public Future<TransactionStatus> commitAsync()
	    throws TransactionStateException {
	if (!threadOwnsTransaction()) {
	    log
		    .error("A thread has tried to commit a transaction it does not own");
	    throw new TransactionalSystemRuntimeException(
		    "Thread does not own the transaction");
	}
	return this.transactionManager.commitAsync();
    }

    /**
     * Checks whether rollback is set to be the only possible outcome of this
     * transaction
     * 
     * @return <code>true</code> if this transaction has to rollback no matter
     *         what
     */
    @Override
    public boolean isRollbackOnly() {
	return this.rollBackOnly;
    }

    @Override
    public String toString() {
	return "TazioTransaction [seqNo=" + this.seqNo + ", transactionID="
		+ this.transactionID + "]";
    }

    @Override
    public boolean equals(final Object obj) {
	if (this == obj) {
	    return true;
	}
	if (obj == null) {
	    return false;
	}
	if (getClass() != obj.getClass()) {
	    return false;
	}
	final TazioTransactionImpl other = (TazioTransactionImpl) obj;
	if (this.getTransactionId() == null) {
	    if (other.getTransactionId() != null) {
		return false;
	    }
	} else if (!this.getTransactionId().equals(other.getTransactionId())) {
	    return false;
	}
	return true;
    }

    @Override
    public int hashCode() {
	final int prime = 3121;
	final int result = 1;
	return prime
		* result
		+ ((this.getTransactionId() == null) ? 0 : this
			.getTransactionId().hashCode());
    }

}

/**
 * This class is a concrete implementation of the {@link TazioTransactionImpl}
 * class, implementing the singleton pattern.
 * <p>
 * The instance provided by this class is not a real transactions, but it is
 * only used as a placeholder for null transactions where the value
 * <code>null</code> cannot be used
 * </p>
 * <p>
 * Quite every operation of this class throws an
 * {@link UnsupportedOperationException}
 * </p>
 * 
 * @version 0.0.1 24/ott/2009
 * @author Andrea Reale <andrea.reale@gmail.com>
 * 
 */
class NullTazioTransaction extends TazioTransactionImpl {

    private static final long serialVersionUID = 6155814276255910244L;
    private static NullTazioTransaction instance = null;

    private NullTazioTransaction() {
	super(null, null, TID.NULL_TID);
	this.setStatus(TransactionStatus.NO_TRANSACTION);
	this.setSeqNo(0);
    }

    static NullTazioTransaction getInstance() {
	if (instance == null) {
	    instance = new NullTazioTransaction();
	}

	return instance;
    }

    /**
     * <strong>Operation not supported</strong>
     * 
     * @throws UnsupportedOperationException
     */
    @Override
    public void commit() {
	throw new UnsupportedOperationException();
    }

    /**
     * <strong>Operation not supported</strong>
     * 
     * @throws UnsupportedOperationException
     */
    @Override
    public Future<TransactionStatus> commitAsync() {
	throw new UnsupportedOperationException();
    }

    /**
     * <strong>Operation not supported</strong>
     * 
     * @throws UnsupportedOperationException
     */
    @Override
    public void rollback() {
	throw new UnsupportedOperationException();
    }

    /**
     * <strong>Operation not supported</strong>
     * 
     * @throws UnsupportedOperationException
     */
    @Override
    public void setRollbackOnly() {
	throw new UnsupportedOperationException();
    }

    /**
     * NullTransactions are equals if they have the same NULL TID. Actually no
     * transaction other then NullTazioTransaction should have a NULL_TID
     */
    @Override
    public boolean equals(final Object obj) {
	if (this == obj) {
	    return true;
	}
	if (obj == null) {
	    return false;
	}
	if (getClass() != obj.getClass()) {
	    return false;
	}
	final NullTazioTransaction other = (NullTazioTransaction) obj;
	if (this.getTransactionId() == null) {
	    if (other.getTransactionId() != null) {
		return false;
	    }
	} else if (!this.getTransactionId().equals(other.getTransactionId())) {
	    return false;
	}
	return true;
    }

    @Override
    public int hashCode() {
	final int prime = 3121;
	final int result = 1;
	return prime
		* result
		+ ((this.getTransactionId() == null) ? 0 : this
			.getTransactionId().hashCode());
    }

    @Override
    public String toString() {
	return "NULL_TRANSACTION";
    }
}
